from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent  # 尝试使用 WriteRowsEvent
)
import re
from elasticsearch import Elasticsearch


def main():
    es = Elasticsearch(['http://39.107.248.141:9200'])
    stream = BinLogStreamReader(
        connection_settings={
            "host": "localhost",
            "port": 3306,
            "user": "root",
            "passwd": "root"
        },
        only_events=[WriteRowsEvent, DeleteRowsEvent, UpdateRowsEvent],
        server_id=100
    )

    table_pattern = re.compile(r'(users|news)')

    try:
        for binlogevent in stream:
            if isinstance(binlogevent, DeleteRowsEvent):
                for row in binlogevent.rows:
                    table = binlogevent.table
                    if table_pattern.match(table):
                        print(f"Detected DELETE on table {table}")
                        es.index(
                            index="binlog_index",
                            body={
                                "event_type": "DELETE",
                                "table": table,
                                "row": row,
                                "timestamp": binlogevent.timestamp
                            }
                        )
            elif isinstance(binlogevent, UpdateRowsEvent):
                for row in binlogevent.rows:
                    table = binlogevent.table
                    if table_pattern.match(table):
                        print(f"Detected UPDATE on table {table}")
                        es.index(
                            index="binlog_index",
                            body={
                                "event_type": "UPDATE",
                                "table": table,
                                "row": row,
                                "timestamp": binlogevent.timestamp
                            }
                        )
            elif isinstance(binlogevent, WriteRowsEvent):
                for row in binlogevent.rows:
                    table = binlogevent.table
                    if table_pattern.match(table):
                        print(f"Detected INSERT on table {table}")
                        es.index(
                            index="binlog_index",
                            body={
                                "event_type": "INSERT",
                                "table": table,
                                "row": row,
                                "timestamp": binlogevent.timestamp
                            }
                        )
    except Exception as e:
        print(f"Error occurred: {e}")
    finally:
        stream.close()


if __name__ == "__main__":
    main()